package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo14mapPartition {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("mapPartition算子演示")
    val context = new SparkContext(conf)
    //====================================================
    //需求：统计总分年级排名前10的学生的各科分数
    //读取分数文件数据
    val scoreRDD: RDD[String] = context.textFile("spark/data/ws/*") // 读取数据文件

    println(scoreRDD.getNumPartitions)

    /**
     *  mapPartition: 主要作用是一次处理一个分区的数据，将一个分区的数据一个一个传给后面的函数进行处理
     *
     *  迭代器中存放的是一个分区的数据
     */
//    val mapPartitionRDD: RDD[String] = scoreRDD.mapPartitions((itr: Iterator[String]) => {
//
//      println(s"====================当前处理的分区====================")
//      //这里写的逻辑是作用在一个分区上的所有数据
//      val words: Iterator[String] = itr.flatMap(_.split("\\|"))
//      words
//    })

//    mapPartitionRDD.foreach(println)

    scoreRDD.mapPartitionsWithIndex{
      case (index:Int,itr: Iterator[String]) =>
        println(s"当前所处理的分区编号是:${index}")
        itr.flatMap(_.split("\\|"))
    }.foreach(println)

  }

}
